package com.we.flink.userlabelplatform.tag;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.we.flink.userlabelplatform.table.TableSaveMoneyCardMember;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Locale;

public class TagSaveMoneyCardMember {
    public long uid;
    public Integer save_money_card_open_time;
    public Integer save_money_card_expired_time;
    public Integer haole_card_open_time;
    public Integer haole_card_expired_time;
    public Integer haojie_card_open_time;
    public Integer haojie_card_expired_time;
    public Integer baoguo_card_open_time;
    public Integer baoguo_card_expired_time;
    public Integer tiecard_expired_time;
    public Integer biguo_card_open_time;
    public Integer save_money_card_update_time;
    public Integer haole_card_update_time;
    public Integer haojie_card_update_time;
    public Integer baoguo_card_update_time;
    public Integer biguo_card_update_time;

    public static Logger LOG = LoggerFactory.getLogger(TagSaveMoneyCardMember.class);

    public TagSaveMoneyCardMember() {}

    public static TagSaveMoneyCardMember init(long uid) {
        return new TagSaveMoneyCardMember(uid, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0);
    }

    public TagSaveMoneyCardMember(
            long uid,
            Integer save_money_card_open_time,
            Integer save_money_card_expired_time,
            Integer haole_card_open_time,
            Integer haole_card_expired_time,
            Integer haojie_card_open_time,
            Integer haojie_card_expired_time,
            Integer baoguo_card_open_time,
            Integer baoguo_card_expired_time,
            Integer tiecard_expired_time,
            Integer biguo_card_open_time,
            Integer save_money_card_update_time,
            Integer haole_card_update_time,
            Integer haojie_card_update_time,
            Integer baoguo_card_update_time,
            Integer biguo_card_update_time) {
        this.uid = uid;
        this.save_money_card_open_time = save_money_card_open_time;
        this.save_money_card_expired_time = save_money_card_expired_time;
        this.haole_card_open_time = haole_card_open_time;
        this.haole_card_expired_time = haole_card_expired_time;
        this.haojie_card_open_time = haojie_card_open_time;
        this.haojie_card_expired_time = haojie_card_expired_time;
        this.baoguo_card_open_time = baoguo_card_open_time;
        this.baoguo_card_expired_time = baoguo_card_expired_time;
        this.tiecard_expired_time = tiecard_expired_time;
        this.biguo_card_open_time = biguo_card_open_time;
        this.save_money_card_update_time = save_money_card_update_time;
        this.haole_card_update_time = haole_card_update_time;
        this.haojie_card_update_time = haojie_card_update_time;
        this.baoguo_card_update_time = baoguo_card_update_time;
        this.biguo_card_update_time = biguo_card_update_time;
    }

    public long getUid() {
        return uid;
    }

    public Integer getSave_money_card_open_time() {
        return save_money_card_open_time;
    }

    public Integer getHaole_card_open_time() {
        return haole_card_open_time;
    }

    public Integer getHaojie_card_open_time() {
        return haojie_card_open_time;
    }

    public Integer getSave_money_card_expired_time() {
        return save_money_card_expired_time;
    }

    public Integer getHaole_card_expired_time() {
        return haole_card_expired_time;
    }

    public Integer getHaojie_card_expired_time() {
        return haojie_card_expired_time;
    }

    public Integer getTiecard_expired_time() {
        return tiecard_expired_time;
    }

    public Integer getBaoguo_card_open_time() {
        return baoguo_card_open_time;
    }

    public Integer getBaoguo_card_expired_time() {
        return baoguo_card_expired_time;
    }

    public Integer getBiguo_card_open_time() {
        return biguo_card_open_time;
    }

    public Integer getSave_money_card_update_time() {
        return save_money_card_update_time;
    }

    public Integer getHaole_card_update_time() {
        return haole_card_update_time;
    }

    public Integer getHaojie_card_update_time() {
        return haojie_card_update_time;
    }

    public Integer getBaoguo_card_update_time() {
        return baoguo_card_update_time;
    }

    public Integer getBiguo_card_update_time() {
        return biguo_card_update_time;
    }

    @Override
    public String toString() {
        SimpleDateFormat simpleDateFormat =
                new SimpleDateFormat("yyyy-MM-dd HH:mm:ss", Locale.CHINA);
        /** just for confirm data */
        return "{"
                + "uid="
                + uid
                + ", save_money_card_exp_time="
                + simpleDateFormat.format(new Date(save_money_card_expired_time * 1000L))
                + " seconds: "
                + save_money_card_expired_time
                + ", haole_card_exp_time="
                + simpleDateFormat.format(new Date(haole_card_expired_time * 1000L))
                + " seconds: "
                + haole_card_expired_time
                + ", haojie_card_exp_time="
                + simpleDateFormat.format(new Date(haojie_card_expired_time * 1000L))
                + " seconds: "
                + haojie_card_expired_time
                + '}';
    }

    public static SingleOutputStreamOperator<String> getFlatMapStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        Table tableRes = tableEnv.sqlQuery(TableSaveMoneyCardMember.QUERYSQL);
        DataStream<Tuple2<Boolean, TableSaveMoneyCardMember>> inputStream =
                tableEnv.toRetractStream(tableRes, TableSaveMoneyCardMember.class);

        SingleOutputStreamOperator<String> out =
                inputStream
                        .filter(
                                new FilterFunction<Tuple2<Boolean, TableSaveMoneyCardMember>>() {
                                    @Override
                                    public boolean filter(
                                            Tuple2<Boolean, TableSaveMoneyCardMember> value)
                                            throws Exception {
                                        return value.f0;
                                    }
                                })
                        .uid("TagSaveMonCardMember-filter")
                        .keyBy(
                                new KeySelector<Tuple2<Boolean, TableSaveMoneyCardMember>, Long>() {
                                    @Override
                                    public Long getKey(
                                            Tuple2<Boolean, TableSaveMoneyCardMember> value)
                                            throws Exception {
                                        return value.f1.getUid();
                                    }
                                })
                        .flatMap(new SaveMoneyCardFlatMapFunciton());
        return out;
    }

    private static class SaveMoneyCardFlatMapFunciton
            extends RichFlatMapFunction<Tuple2<Boolean, TableSaveMoneyCardMember>, String> {
        ValueState<TagSaveMoneyCardMember> saveMoneyCardMemberTagVState;

        @Override
        public void open(Configuration parameters) throws Exception {
            saveMoneyCardMemberTagVState =
                    getRuntimeContext()
                            .getState(
                                    new ValueStateDescriptor<TagSaveMoneyCardMember>(
                                            "SaveMoneyCardMemberTagVState",
                                            TagSaveMoneyCardMember.class));
        }

        @Override
        public void flatMap(Tuple2<Boolean, TableSaveMoneyCardMember> value, Collector<String> out)
                throws Exception {
            long uid = value.f1.getUid();
            int update_time = (int) (value.f1.getUpdate_time().getTime() / 1000);
            int open_time = (int) (value.f1.getOpen_time().getTime() / 1000);
            int end_time = (int) (value.f1.getEnd_time().getTime() / 1000);

            TagSaveMoneyCardMember tmpState = saveMoneyCardMemberTagVState.value();
            TagSaveMoneyCardMember curState =
                    tmpState == null ? TagSaveMoneyCardMember.init(uid) : tmpState;
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("uid", uid);

            /** 默认不为null 且 有更新时才触发 sink kafka 目标 缩减数据量 */
            switch (value.f1.getCard_type()) {
                case 1:
                    if (update_time > curState.getSave_money_card_update_time()
                            && (curState.save_money_card_open_time != open_time
                                    || curState.save_money_card_expired_time != end_time)) {
                        curState.save_money_card_open_time = open_time;
                        curState.save_money_card_expired_time = end_time;
                        curState.save_money_card_update_time = update_time;
                        jsonObject.put(
                                "save_money_card_open_time", curState.save_money_card_open_time);
                        jsonObject.put(
                                "save_money_card_expired_time",
                                curState.save_money_card_expired_time);
                        out.collect(jsonObject.toString());
                        saveMoneyCardMemberTagVState.update(curState);
                    }
                    break;
                case 10:
                    if (update_time > curState.getHaole_card_update_time()
                            && (curState.haole_card_open_time != open_time
                                    || curState.haole_card_expired_time != end_time)) {
                        curState.haole_card_open_time = open_time;
                        curState.haole_card_expired_time = end_time;
                        curState.haole_card_update_time = update_time;
                        jsonObject.put("haole_card_open_time", curState.haole_card_open_time);
                        jsonObject.put("haole_card_expired_time", curState.haole_card_expired_time);
                        out.collect(jsonObject.toString());
                        saveMoneyCardMemberTagVState.update(curState);
                    }
                    break;
                case 12:
                    if (update_time > curState.getHaojie_card_update_time()
                            && (curState.haojie_card_open_time != open_time
                                    || curState.haojie_card_expired_time != end_time)) {
                        curState.haojie_card_open_time = open_time;
                        curState.haojie_card_expired_time = end_time;
                        curState.haojie_card_update_time = update_time;
                        jsonObject.put("haojie_card_open_time", curState.haojie_card_open_time);
                        jsonObject.put(
                                "haojie_card_expired_time", curState.haojie_card_expired_time);
                        out.collect(jsonObject.toString());
                        saveMoneyCardMemberTagVState.update(curState);
                    }
                    break;
                case 19:
                    if (update_time > curState.getBaoguo_card_update_time()
                            && (curState.baoguo_card_open_time != open_time
                                    || curState.baoguo_card_expired_time != end_time)) {
                        curState.baoguo_card_open_time = open_time;
                        curState.baoguo_card_expired_time = end_time;
                        curState.baoguo_card_update_time = update_time;
                        jsonObject.put("baoguo_card_open_time", curState.baoguo_card_open_time);
                        jsonObject.put(
                                "baoguo_card_expired_time", curState.baoguo_card_expired_time);
                        out.collect(jsonObject.toString());
                        saveMoneyCardMemberTagVState.update(curState);
                    }
                    break;
                case 24:
                    if (update_time > curState.getBiguo_card_update_time()
                            && curState.biguo_card_open_time != open_time) {
                        curState.biguo_card_open_time = open_time;
                        curState.biguo_card_update_time = update_time;
                        jsonObject.put("biguo_card_open_time", curState.biguo_card_open_time);
                        out.collect(jsonObject.toString());
                        saveMoneyCardMemberTagVState.update(curState);
                    }
                    break;
                default:
                    LOG.error("value.getCard_type : " + value.f1.getCard_type());
                    break;
            }
        }
    }
}
